package tag


import java.sql.DriverManager

import Configer.Config
import ch.hsr.geohash.GeoHash
import org.apache.commons.lang.StringUtils
import org.apache.spark.sql.SQLContext
import org.apache.spark.{SparkConf, SparkContext}
import scalikejdbc._
import util.GetUUID

//为原始数据打标签
object DMPTags {
  def main(args: Array[String]): Unit = {
    //sparkContext
    val conf = new SparkConf()
    conf.setMaster("local[1]")
    conf.setAppName(s"${this.getClass.getName}")
    conf.set("spark.serializer", Config.serializer)

    val sc = new SparkContext(conf)

    val sQLContext = new SQLContext(sc)

    //读数据
    val dataFrame = sQLContext.read.parquet(Config.parquetPath)
    val dictLines = sc.textFile("C:\\Users\\44323\\Desktop\\资料PDF\\app_dict.txt")
    val stopLines = sc.textFile("C:\\Users\\44323\\Desktop\\资料PDF\\stopwords.txt")


    //广播变量
    val dictMap = dictLines.map(_.split("\t", -1)).filter(_.length >= 5).map(arr => (arr(4), arr(1))).collectAsMap()
    val dictBc = sc.broadcast(dictMap)

    val stopMap = stopLines.map((_, null)).collectAsMap()
    val stopBc = sc.broadcast(stopMap)


    //分析数据
    dataFrame.filter(
      """
        |imei != "" or mac != "" or idfa != "" or openudid != "" or androidid != "" or
        |imeimd5 != "" or macmd5 != "" or idfamd5 != "" or openudidmd5 != "" or androididmd5 != "" or
        |imeisha1 != "" or macsha1 != "" or idfasha1 != "" or openudidsha1 != "" or androididsha1 != ""
      """.stripMargin)
      .mapPartitions(partitio => {

        val presutl = partitio.map(
          row => {
            var list = List[(String, Int)]()

            //广告位
            val listAd: List[(String, Int)] = AdTags.makeTags(row)

            //app
            val listApp: List[(String, Int)] = AppTags.makeTags(row, dictBc)

            //渠道
            val listChan = ChannelTags.makeTags(row)

            //设备  操作系统  联网方式  运营商
            val listDev = DeviceTags.makeTags(row)
            //关键字
            val listKey = KeyTags.makeTags(row, stopBc)

            //地域标签
            val listArea = AreaTags.makeTags(row)

            //商圈
            val conn = DriverManager.getConnection(Config.url, Config.user, Config.password)
            val listBusiness = BusinessTags.makeTags(row, conn)
            conn.close()

            val userid = GetUUID.getUUID(row)
            (userid, listAd ++ listApp ++ listChan ++ listDev ++ listKey++listBusiness)
          })

        presutl
      }).reduceByKey {
      (list1, list2) =>
        (list1 ++ list2).groupBy(_._1) //.mapValues(v => v.length).toList
          //  .mapValues(v=>v.map(_._2).sum).toList
          .mapValues(v => v.foldLeft(0)(_ + _._2)).toList
    }.coalesce(1).saveAsTextFile("C:\\Users\\44323\\Desktop\\资料PDF\\tags")
    sc.stop()

  }
}
